package com.example.message_queue.mqserver.core;

import com.example.message_queue.common.Consumer;
import com.example.message_queue.common.ConsumerEnv;
import com.example.message_queue.common.MqException;
import com.example.message_queue.mqserver.VirtualHost;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * Created with Intellij IDEA.
 * Description;
 * User:TQ02
 * Data:2024-04-04
 * Time:11:29
 * Text：
 */
public class ConsumerManager {
    private VirtualHost parent;

    //持有一个线程池，负责执行具体的回调任务
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);
    //存放令牌的队列,存储队列名
    private BlockingDeque<String> tokenQueue = new LinkedBlockingDeque<>();
    //扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost p){
        parent = p;
        scannerThread = new Thread(() ->{
            try{
                //拿令牌
                String queuename = tokenQueue.take();
                //根据令牌，找队列
                MSGQueue queue = parent.getMemoryDataCenter().getQueue(queuename);
                if(queue == null){
                    throw new MqException("[ConsumerManger]ConsumerManger:取令牌发现，该队列名不存在"+queuename);
                }
                //从该队列中消费消息
                synchronized (queue){
                    consumeMessage(queue);
                }
            } catch (InterruptedException | MqException e) {
                throw new RuntimeException(e);
            }
        });
        //线程设为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }
    public void notifyConsume(String name) throws InterruptedException {
        tokenQueue.put(name);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        //找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getQueue(queueName);
        if(queue == null){
            throw new MqException("[ConsumerManager]队列不存在！queueName"+ queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue){
            queue.addConsumerEnv(consumerEnv);
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for(int i=0; i < n; i++){
                //这个方法调用一次就消费一条消息
                consumeMessage(queue);
            }
        }

    }
    //消费队列的新消息
    private void consumeMessage(MSGQueue queue) {
        //按照轮询的方式，找消费者出来
        ConsumerEnv luckDog = queue.chooseConsumer();
        if(luckDog == null){
            return;
        }
        //从队列中取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message == null){
            return;
        }
        workerPool.submit(()->{
            try{
                // 1. 把消息放到待确认的集合中. 这个操作势必在执行回调之前.
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                // 2. 真正执行回调操作
                luckDog.getConsumer().handleDelivery(luckDog.getConsumerTag(),message.getBasicProertirs(),
                        message.getBody());
                // 3. 如果当前是 "自动应答" , 就可以直接把消息删除了.
                //    如果当前是 "手动应答" , 则先不处理, 交给后续消费者调用 basicAck 方法来处理.
                if (luckDog.isAutoAck()) {
                    // 1) 删除硬盘上的消息
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    // 2) 删除上面的待确认集合中的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    // 3) 删除内存中消息中心里的消息
                    parent.getMemoryDataCenter().deleteMessage(message.getMessageId());
                    System.out.println("[ConsumerManager] 消息被成功消费! queueName=" + queue.getName());
                    System.out.println("[ConsumerManager]consumeMessage：消息消费成功");
                }
            }catch (Exception e) {
                    e.printStackTrace();
            }
        });
    }

}
